package cn.itcast.flink.cdc;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *  * 基于DataStream实现Flink CDC案例演示：从MySQL数据库实时获取数据，打印控制台。
 * @author lilulu
 * @date 2023-04-10 20:50
 */
public class FlinkCdcDataStreamDemo {
    @SneakyThrows
    public static void main(String[] args) {
        //执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(3000);
        env.setParallelism(1);

        //数据源
//        MySQL-CDC-Source
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
                .hostname("node1")
                .port(3306).username("root")
                .password("123456")
                .databaseList("db_flink") // set captured database
                .tableList("db_flink.tbl_users") // set captured table
                .startupOptions(StartupOptions.initial())
                // 表示第1次读取表中数据，先全量加载，后增量加载
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        //添加数据源
        DataStreamSource<String> logStream = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source");
        logStream.printToErr();
        env.execute("FlinkCDDataStreamDemo");

    }
}
